package app.dwd;

import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.SQLUtil;

public class dwd_video_play extends BaseSQLApp {
    public static void main(String[] args) {
        new dwd_video_play().init(
                2001,
                2,
                "Dwd_Video_Play"
        );
    }
    //播放事实表
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_Video_Play");
//        user_chapter_process
        Table userplay = tEnv.sqlQuery("select data['id'] id,  " +
                "       data['course_id'] course_id,  " +
                "       data['chapter_id'] chapter_id,  " +
                "       data['user_id'] user_id,  " +
                "       data['position_sec'] position_sec,  " +
                "       data['create_time'] create_time,  " +
                "       data['update_time'] update_time,  " +
                "       ts  " +
                "from ods_db  " +
                "where `database`='edu'   " +
                "and `table`='user_chapter_process'  " +
                "and data['deleted']='0'   " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("userplay",userplay);
//        video_info
        Table video = tEnv.sqlQuery("select data['id']  id,  " +
                "       data['video_name']  video_name,  " +
                "       data['during_sec']  during_sec,  " +
                "       data['video_status']  video_status,  " +
                "       data['video_size']  video_size,  " +
                "       data['video_url']  video_url,  " +
                "       data['video_source_id']  video_source_id,  " +
                "       data['version_id']  version_id,  " +
                "       data['chapter_id']  chapter_id,  " +
                "       data['course_id']  course_id,  " +
                "       data['publisher_id']  publisher_id,  " +
                "       data['create_time']  create_time,  " +
                "       data['update_time']  update_time  " +
                "from ods_db  " +
                "where `database`='edu'  " +
                "  and `table`='video_info'  " +
                "  and data['deleted']='0'  " +
                "  and `type`='insert'");
        tEnv.createTemporaryView("video",video);
//        join
        Table result = tEnv.sqlQuery("select " +
                "   up.id, " +
                "   up.course_id, " +
                "   up.user_id," +
                "   vd.video_name," +
                "   vd.video_source_id," +
                "   vd.version_id," +
                "   vd.chapter_id," +
                "   vd.publisher_id," +
                "   up.position_sec user_position_sec," +
                "   vd.during_sec video_during_sec," +
                "   ts," +
                "   current_row_timestamp() row_op_ts" +
                "   from userplay up left join video  vd" +
                "   on up.chapter_id=vd.chapter_id");
        tEnv.executeSql("create table dwd_video_play(" +
                "id string,  " +
                "course_id string,  " +
                "user_id string,  " +
                "video_name string,  " +
                "video_source_id string,  " +
                "version_id string,  " +
                "chapter_id string,  " +
                "publisher_id string,  " +
                "user_position_sec string,  " +
                "video_during_sec string,  " +
                "ts bigint,  " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(id) not enforced " +
                ")"+ SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_VIDEO_PLAY));

        result.executeInsert("dwd_video_play");





    }
}
